package cn.jly.flink.table;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author hxfz
 * @date 2020/7/22
 */
public class FirstTableApiApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings setting = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment streamTableEnv = StreamTableEnvironment.create(env, setting);

        SingleOutputStreamOperator<Person> personDataStream = env.fromElements(
                Tuple2.of(1, "tom"),
                Tuple2.of(2, "Bom"),
                Tuple2.of(3, "Kobe"),
                Tuple2.of(4, "James")
        ).map(new MapFunction<Tuple2<Integer, String>, Person>() {
            @Override
            public Person map(Tuple2<Integer, String> value) throws Exception {
                return Person.builder().id(value.f0).name(value.f1).build();
            }
        });

        /*personDataStream.print();*/
        // 注册table
        streamTableEnv.registerDataStream("personTable", personDataStream, "id, name");

        // table api
        /*Table table = streamTableEnv.scan("personTable")
                .where("like(name, '%o%')")
                .select("name");*/

        // SQL
        Table table = streamTableEnv.sqlQuery("select name from personTable where name like '%o%'");

        // table to dataStream
        DataStream<Row> resultDataStream = streamTableEnv.toAppendStream(table, Row.class);
        resultDataStream.print();

        streamTableEnv.execute("FirstTableApiApp");
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Builder
    public static class Person {
        private Integer id;
        private String name;
    }
}
